package mspbots.smileback.client.core.reviews;

import com.fasterxml.jackson.databind.JsonNode;
import lombok.extern.log4j.Log4j2;
import mspbots.smileback.client.core.AbstractToolsUtil;
import mspbots.smileback.client.core.BaseSyncService;
import mspbots.smileback.common.*;
import org.springframework.stereotype.Service;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.time.LocalDateTime;

/**
 * mspbots.smileback.client.core.reviews.ReviewsServiceImpl
 *
 * @author Alex bob(https://github.com/vnobo)
 * @date Created by 2020/4/14
 */
@Log4j2
@Service
public class ReviewsServiceImpl extends AbstractToolsUtil implements BaseSyncService {

    @Override
    public Mono<Void> syncFast(Long tenantId) {
        ResultSync resultSync = ResultSync.withDefault(tenantId, "REVIEWS-SYNC");
        return this.filterCacheGet(tenantId, resultSync)
                .doOnNext(resultNode -> this.sendEntity(RabbitKeys.REVIEWS_SYNC_QUEUE_KEY,ResultMsg.builder()
                        .requestId(resultSync.getRequestId()).id(resultNode.get("id").longValue())
                        .action("none").entity(resultNode).tenantId(tenantId)
                        .syncDate(LocalDateTime.now()).build()))
                .doOnNext(resultNode->super.recordAnchorTime(RedisKeys.REVIEWS_ANCHOR_REDIS_KEY_PREFIX + tenantId,
                        resultNode.get("last_modified").textValue()))
                .then();
    }

    private Flux<JsonNode> filterCacheGet(Long tenantId, ResultSync resultSync) {
        String anchorTimeKey =RedisKeys.REVIEWS_ANCHOR_REDIS_KEY_PREFIX + tenantId;
        return this.readAnchorTime(anchorTimeKey)
                .map(time -> QueryParams.withDefault().setModifiedSince(time))
                .flatMapMany(queryParams -> this.fetch(queryParams)
                        .subscriberContext(ctx -> ctx.put("resultSync", resultSync.setParams(queryParams))))
                .subscriberContext(ctx -> ctx.put("tenantId", tenantId))
                .flatMap(jsonNode -> Flux.fromIterable(jsonNode.withArray("results")));
    }

    private Flux<JsonNode> fetch(QueryParams queryParams) {
        return this.smileBackClient.fetch(this.smileBackClient.buildUrl("/api/v3/reviews/", queryParams));
    }

}
